package com.jml.netty.Mq简单原理;

import com.alibaba.fastjson.JSONObject;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import org.apache.commons.lang3.StringUtils;

import java.io.UnsupportedEncodingException;
import java.util.ArrayList;
import java.util.concurrent.LinkedBlockingQueue;


public class NettyMQServer {
    public void bind(int port) throws Exception {
        /**
         * Netty 抽象出两组线程池BossGroup和WorkerGroup
         * BossGroup专门负责接收客户端的连接, WorkerGroup专门负责网络的读写。
         */
        EventLoopGroup bossGroup = new NioEventLoopGroup();
        EventLoopGroup workerGroup = new NioEventLoopGroup();
        ServerBootstrap bootstrap = new ServerBootstrap();
        try {
            bootstrap.group(bossGroup, workerGroup)
                    // 设定NioServerSocketChannel 为服务器端
                    .channel(NioServerSocketChannel.class)
                    //BACKLOG用于构造服务端套接字ServerSocket对象，标识当服务器请求处理线程全满时，
                    //用于临时存放已完成三次握手的请求的队列的最大长度。如果未设置或所设置的值小于1，Java将使用默认值50。
                    .option(ChannelOption.SO_BACKLOG, 100)
                    // 服务器端监听数据回调Handler
                    .childHandler(new ChildChannelHandler());
            //绑定端口, 同步等待成功;
            ChannelFuture future = bootstrap.bind(port).sync();
            System.out.println("当前服务器端启动成功...");
            //等待服务端监听端口关闭
            future.channel().closeFuture().sync();
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            //优雅关闭 线程组
            bossGroup.shutdownGracefully();
            workerGroup.shutdownGracefully();
        }
    }

    private class ChildChannelHandler extends ChannelInitializer<SocketChannel> {
        @Override
        protected void initChannel(SocketChannel ch) throws Exception {
            // 设置异步回调监听
            ch.pipeline().addLast(new JmlServerHandler());
//////            1. 演示LineBasedFrameDecoder编码器
//            ch.pipeline().addLast(new LineBasedFrameDecoder(1024));
//            ch.pipeline().addLast(new StringDecoder());
        }
    }

    public static void main(String[] args) throws Exception {
        int port = 9008;
        new NettyMQServer().bind(port);
    }


    private static final String type_consumer = "consumer";

    private static final String type_producer = "producer";
    /**
     * mq缓存消息
     */
    private static LinkedBlockingQueue<String> msgs = new LinkedBlockingQueue<String>();
    /*
     *存放消费者的长连接信息，channel通道连接
     */
    private static ArrayList<ChannelHandlerContext> consumerChannels = new ArrayList();

    // 生产者投递消息的：topicName
    public class JmlServerHandler extends SimpleChannelInboundHandler<Object> {
        /**
         * 服务器接收客户端请求
         *
         * @param ctx
         * @param data
         * @throws Exception
         */
        @Override
        protected void channelRead0(ChannelHandlerContext ctx, Object data)
                throws Exception {
            String body = byteBufToString(data);
            JSONObject jsonObject = JSONObject.parseObject(body);
            String type = jsonObject.getString("type");
            switch (type) {
                case type_consumer:
                    consumer(ctx);
                    return;
                case type_producer:
                    String msg = jsonObject.getString("msg");
                    producer(msg);
                    return;
            }

        }


        private void consumer(ChannelHandlerContext ctx) {
            // 保存连接
            consumerChannels.add(ctx);
            String poll = msgs.poll();
            if (!StringUtils.isEmpty(poll)) {
                // 从MQ服务器端获取消息给消费者
                ByteBuf resp = Unpooled.copiedBuffer(poll.getBytes());
                ctx.writeAndFlush(resp);
            }
            // 思路

        }

        /**
         * 缓存生产者投递消息
         *
         * @param msg
         */
        private void producer(String msg) {
            // 缓存消息
            msgs.offer(msg);
            //把消息推送给消费者
            consumerChannels.forEach((ctx) -> {
                String poll = msgs.poll();
                if (StringUtils.isEmpty(poll)) {
                    return;
                }
                // 发送数据给消费者
                ByteBuf resp = Unpooled.copiedBuffer(poll.getBytes());
                ctx.writeAndFlush(resp);
            });
            // 分组消息

        }

        private String byteBufToString(Object msg) throws UnsupportedEncodingException {
            if (msg == null) {
                return null;
            }
            ByteBuf buf = (ByteBuf) msg;
            byte[] req = new byte[buf.readableBytes()];
            buf.readBytes(req);
            String body = new String(req, "UTF-8");
            return body;
        }

        @Override
        public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
            ctx.flush();
        }

        //断开连接调这个方法
        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause)
                throws Exception {
            consumerChannels.remove(ctx);
            ctx.close();
        }
    }
}
